package com.innovation.wjrpc.communication;

import com.innovation.wjrpc.protocol.Peer;
import com.innovation.wjrpc.transport.TransportClient;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.ByteToMessageDecoder;
import io.netty.handler.codec.MessageToByteEncoder;
import io.netty.handler.codec.json.JsonObjectDecoder;
import lombok.Data;

import java.net.InetSocketAddress;

/**
 * @author innoyiya
 * @date 2021/9/8 10:24
 */
@Data
public class NettyTransportClient implements TransportClient<ChannelFuture, ByteBuf> {
    private ChannelFuture channelFuture;
    private EventLoopGroup group;
    private ChannelHandlerContext cxt;
    private ChannelHandler handler;
    private ByteToMessageDecoder byteToMessageDecoder;
    private MessageToByteEncoder messageToMessageDecoder;

    @Override
    public void connect(Peer peer) {
        group = new NioEventLoopGroup();
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(group)
                .channel(NioSocketChannel.class)
                .remoteAddress(new InetSocketAddress(peer.getHost(), peer.getPort()))
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) {
                        ch.pipeline().addLast(new JsonObjectDecoder());
                        ch.pipeline().addLast(handler);
                    }
                });

        try {
            channelFuture = bootstrap.connect().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    @Override
    public ChannelFuture write(ByteBuf data) {
        return channelFuture.channel().writeAndFlush(data).addListener(future -> {
            System.out.println("send data");
            System.out.println(data.readableBytes());
        });
    }

    @Override
    public void close() {
        try {
            channelFuture.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }finally {
            try {
                group.shutdownGracefully().sync();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}
